package com.briup.searchengine.handle;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

/**
 * @author adam
 * @date 2022/6/13
 */
public class Step6_HoldJoinResult extends Configured implements Tool {

    public static class HoldJoinResultMapper extends TableMapper<ImmutableBytesWritable, MapWritable> {
        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
            NavigableMap<byte[], byte[]> nvm = value.getFamilyMap("page".getBytes());
            MapWritable map = new MapWritable();
            if (nvm.size() == 4) {
                byte[] rank = value.getValue("page".getBytes(), "rank".getBytes());
                byte[] keyword = value.getValue("page".getBytes(), "key".getBytes());
                byte[] title = value.getValue("page".getBytes(), "title".getBytes());
                byte[] content = value.getValue("page".getBytes(), "content".getBytes());
                System.out.println(Bytes.toFloat("rank".getBytes()));
                map.put(new Text("rank"), new BytesWritable(rank));
                map.put(new Text("key"), new BytesWritable(keyword));
                map.put(new Text("title"), new BytesWritable(title));
                map.put(new Text("content"), new BytesWritable(content));
                context.write(key, map);
            }
        }
    }

    public static class HoldJoinResultReducer extends TableReducer<ImmutableBytesWritable, MapWritable, NullWritable> {
        @Override
        protected void reduce(ImmutableBytesWritable key, Iterable<MapWritable> values, Context context) throws IOException, InterruptedException {
            Put put = new Put(key.get(), key.getOffset(), key.getLength());
            for (MapWritable value : values) {
                Set<Map.Entry<Writable, Writable>> entrySet = value.entrySet();
                for (Map.Entry<Writable, Writable> entry : entrySet) {

                    BytesWritable v = (BytesWritable) entry.getValue();
                    put.addColumn("page".getBytes(), entry.getKey().toString().getBytes(), v.getBytes());
                }
            }
            context.write(NullWritable.get(), put);
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        //conf.set("hbase.zookeeper.quorum", "hadoop01:2181");
        Job job = Job.getInstance(conf, "HoldJoinResult");
        job.setJarByClass(this.getClass());
        TableMapReduceUtil.initTableMapperJob("last_result", new Scan(), HoldJoinResultMapper.class, ImmutableBytesWritable.class, MapWritable.class, job);
        TableMapReduceUtil.initTableReducerJob("join_result", HoldJoinResultReducer.class, job);
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Step6_HoldJoinResult(), args);
    }
}
